package base

import java.util.Properties
import bean.HBaseMeta
import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

trait BaseModel {
  //一样的需求步骤写成普通变量和方法  不一样的需求步骤使用抽象方法和抽象变量重写

  //程序名称
  def sparkName(): String

  //读取四五级标签所用的 id
  def getId: Int

  //读取配置文件
  private val conf: Config = ConfigFactory.load()
  private val url: String = conf.getString("jdbc.mysql.url")
  private val tableName: String = conf.getString("jdbc.mysql.tableName")

  // 创建sparkSession
  val spark: SparkSession = SparkSession.builder.appName(sparkName()).master("local[*]").getOrCreate
  val sc: SparkContext = spark.sparkContext
  sc.setLogLevel("WARN")

  //隐式转化
  import spark.implicits._
  import org.apache.spark.sql.functions._

  //获取mysql连接
  def getMySQLCoon() = {
    spark.read.jdbc(url, tableName, new Properties())
  }

  //获取四级标签 封装hbaseMeta 为读取HBASE做准备
  def getFourTag(mysqlCoon: DataFrame) = {
    val fourRow: Dataset[Row] = mysqlCoon.select('rule).where("id=" + getId)
    val fourMap: Map[String, String] = fourRow.map(row => {
      //inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,birthday
      row.getAs("rule").toString.split("##")
        //inType=HBase zkHosts=192.168.10.20
        .map(line => {
          //zkHosts 192.168.10.20
          val arr: Array[String] = line.split("=")
          (arr(0), arr(1))
        })
    }).collectAsList().get(0).toMap

    //封装hbaseMeta的方法
    def getHbaseMeta(fourMap: Map[String, String]): HBaseMeta = {
      HBaseMeta(
        fourMap.getOrElse(HBaseMeta.INTYPE, ""),
        fourMap.getOrElse(HBaseMeta.ZKHOSTS, ""),
        fourMap.getOrElse(HBaseMeta.ZKPORT, ""),
        fourMap.getOrElse(HBaseMeta.HBASETABLE, ""),
        fourMap.getOrElse(HBaseMeta.FAMILY, ""),
        fourMap.getOrElse(HBaseMeta.SELECTFIELDS, ""),
        fourMap.getOrElse(HBaseMeta.ROWKEY, "")
      )
    }

    //封装hbaseMeta
    getHbaseMeta(fourMap)

  }

  //获取五级标签 为计算、匹配标签做准备 单独计算
  def getFiveTag(mysqlCoon: DataFrame) = {
    mysqlCoon.select('id, 'rule).where("pid=" + getId)
  }

  //根据四级标签读取 HBASE数据
  def getHBase(hbaseMeta: HBaseMeta)={
    spark.read.format("tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, hbaseMeta.hbaseTable)
      .option(HBaseMeta.FAMILY, hbaseMeta.family)
      .option(HBaseMeta.SELECTFIELDS, hbaseMeta.selectFields)
      .load()
  }

  //根据五级标签数据和hbase数据 计算、匹配标签（抽象方法）
  def getTags(fiveRow:Dataset[Row],hbaseDF:DataFrame):DataFrame

  //将计算完的标签写入到HBASE
  def saveToHbase(ConstellationTagDF: DataFrame,hbaseMeta: HBaseMeta): Unit = {
    //读取HBASE历史数据
    val historyDF: DataFrame = spark.read.format("tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
      .load()

    //整合historyDF 和 ConstellationTagDF
    historyDF.createOrReplaceTempView("historyDF")
    ConstellationTagDF.createOrReplaceTempView("ConstellationDF")

    //创建udf函数对打入标签做添加、去重操作
    spark.udf.register("updateTags",(hTagsId: String, CTagsId: String) => {
      if (hTagsId == "") {
        CTagsId
      } else if (CTagsId == "") {
        hTagsId
      } else if (hTagsId == "" && CTagsId == "") {
        ""
      } else {
        val str: String = hTagsId + "," + CTagsId
        str.split(",").distinct.mkString(",")
      }
    })

    //标签匹配
    val updateTagsDF: DataFrame = spark.sql(
      """
        |SELECT h.userId,updateTags(h.tagsId,c.tagsId) as tagsId
        |FROM historyDF h
        |LEFT JOIN ConstellationDF c
        |ON h.userId=c.userId
        |""".stripMargin).toDF("userId", "tagsId")

    //写入HBASE
    updateTagsDF.write.format("tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS,"userId,tagsId")
      .save()

  }
  //执行流程
  def exec()={
    //获取MySQL连接
    val mysqlCoon: DataFrame = getMySQLCoon()
    //读取四级标签 封装hbaseMeta
    val hbaseMeta: HBaseMeta = getFourTag(mysqlCoon)
    //获取五级标签
    val fiveRow: Dataset[Row] = getFiveTag(mysqlCoon)
    //获取HBase数据
    val HBaseDF: DataFrame = getHBase(hbaseMeta)
    //重写getTags方法 根据五级标签和HBASE数据 计算匹配出标签
    val ConstellationTagDF: DataFrame = getTags(fiveRow, HBaseDF)
    //将更新标签结果写入HBASE
    saveToHbase(ConstellationTagDF,hbaseMeta)
  }
}
